// @Author EthanScriptOn
// @Desc
package rabbitmq_operator

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/identity"
	"gitee.com/fatzeng/srf_switch_expansion_package/connect"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_config"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator/op_example"
)

type RabbitmqConsumeOperator struct {
	*op_example.LoopDetectionMultiChannelDataOperator
	*identity.IdentityInformationCarrier
}

func (r *RabbitmqConsumeOperator) ExecuteWithLoop(ctx *actuator.SrfSwitchBridgeActuatorContext, operatorConfigIdentifier interface{}, structureParam interface{}) (err error) {
	operatorIdentifier, err := r.IdentityInformationCarrier.GenerateUniqueIdentifier()
	if err != nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error generating unique identifier: %v", err))
		return err
	}
	pipeline := r.GetDataPipeline(operatorIdentifier)
	if pipeline == nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error getting data pipeline: %v", err))
		return err
	}
	consumer, err := connect.GetRabbitmqConsumer()
	if err != nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error getting rabbitmq consumer: %v", err))
		return err
	}

	config := operator.GetOperatorConfig(operatorConfigIdentifier)
	if config == nil {
		return fmt.Errorf("config is nil")
	}
	operatorConfig, ok := config.(*extend_config.RabbitmqConsumeOperatorConfig)
	if !ok {
		return fmt.Errorf("config is not *RabbitmqConsumeOperatorConfig")
	}
	consumeChannel, err := consumer.Consume(operatorConfig.Queue, operatorConfig.Consumer, operatorConfig.AutoAck, operatorConfig.Exclusive, operatorConfig.NoLocal, operatorConfig.NoWait, operatorConfig.Args)
	if err != nil {
		return fmt.Errorf("error consuming: %v", err)
	}
	for msg := range consumeChannel {
		pipeline <- msg
	}
	return
}
